/**
 * 
 */
package data.mysql;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import common.Logger;

import data.IDao;
import data.Pagination;
import data.bean.DataBean;
import data.bean.DatabaseField;
import data.exception.DaoException;
import data.exception.InvalidFieldException;

/**
 * The DAO base class supporting the DAO implementation.
 * 
 * @author chuxiaoyuan
 *
 */
public abstract class BaseDaoSupport<Bean extends DataBean> 
				implements IDao<Bean>, IResultsProcessor<List<Bean>> {

	/**
	 * Constructor.
	 */
	public BaseDaoSupport() {
		
		// Set bean instance
		this.setBeanInstance(this.createBean());
		
		// Set transactional
		this.setTransactional(false);
		
		// Initialize logger
		this.logger = new Logger(this.getClass());
	}

	/**
	 * The logger.
	 */
	private Logger logger;
	
	/**
	 * Database connection.
	 */
	private Connection connection = null;

	/* (non-Javadoc)
	 * @see data.mysql.IDao#getConnection()
	 */
	@Override
	public Connection getConnection() throws DaoException {
		
		if (null != this.connection) {
			// Return the current connection
			return this.connection;
		}
		
		// Get a new connection
		this.connection = MySQLDaoFactory.openConnection();
		return this.connection;
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#closeConnection()
	 */
	@Override
	public void closeConnection() throws DaoException {
		
		if (null != this.connection && !this.isTransactional()) {
			try {
				// Close connection
				this.connection.close();
				
				// Reset
				this.connection = null;
				logger.debug("Connection closed.");
				
			} catch (SQLException e) {
				
				String msg = "Failed to close connection!";
				logger.error(msg, e);
				throw new DaoException(msg, e);
			}
		}
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#setConnection(java.sql.Connection)
	 */
	@Override
	public void setConnection(Connection connection) throws DaoException {
		
		// Try to clean up first.
		this.closeConnection();
		
		// Set connection.
		this.connection = connection;
	}
	
	/**
	 * Indicates if this DAO is inside a bigger transaction scope.
	 */
	private boolean transactional;

	/* (non-Javadoc)
	 * @see data.mysql.IDao#isTransactional()
	 */
	@Override
	public boolean isTransactional() {
		return this.transactional;
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#setTransactional(boolean)
	 */
	@Override
	public void setTransactional(boolean transactional) {
		this.transactional = transactional;
	}
	
	/**
	 * Create DataBean instance.
	 */
	protected abstract Bean createBean();

	/**
	 * DataBean instance.
	 */
	private Bean beanInstance = null;
	
	/**
	 * Get bean instance.
	 */
	protected Bean getBeanInstance() {
		return this.beanInstance;
	}
	
	/**
	 * Set bean instance.
	 */
	private void setBeanInstance(Bean beanInstance) {
		this.beanInstance = beanInstance;
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#getTableName()
	 */
	public String getTableName() {
		return this.getBeanInstance().getTableName();
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#getFields()
	 */
	public List<DatabaseField> getFields() {
		return this.getBeanInstance().getFields();
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#getFields(java.util.List<String>)
	 */
	public List<DatabaseField> getFields(List<String> fieldsName) {
		return this.getBeanInstance().getFields(fieldsName);
	}
	
	/* (non-Javadoc)
	 * @see data.mysql.IDao#getField(java.lang.String)
	 */
	public DatabaseField getField(String fieldName) {
		return this.getBeanInstance().getField(fieldName);
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#isTableEntity()
	 */
	public boolean isTableEntity() {
		return this.getBeanInstance().isTableEntity();
	}
	
	/**
	 * DB logging.
	 */
	protected void log(String msg, Exception e) {
		logger.log("[DB] " + msg, e);
	}

	/**
	 * DB logging.
	 */
	protected void log(String msg) {
		this.log(msg, null);
	}
	
	/**
	 * DB debugging.
	 */
	protected void debug(String msg, Exception e) {
		logger.debug("[DB] " + msg, e);
	}

	/**
	 * DB debugging.
	 */
	protected void debug(String msg) {
		this.debug(msg, null);
	}
	
	/**
	 * DB error log.
	 */
	protected void error(String msg, Exception e) {
		logger.error("[DB] " + msg, e);
	}

	/**
	 * DB error log.
	 */
	protected void error(String msg) {
		this.error(msg, null);
	}
	
	/* (non-Javadoc)
	 * @see data.mysql.IDao#insert(data.bean.DataBean)
	 */
	@Override
	public int insert(Bean bean) throws DaoException {

		if (!this.isTableEntity()) {
			String msg = 
					"Please override this method to implement non table operation.";
			this.error(msg);
			throw new DaoException(msg);
		}
		
		// Get connection
		Connection conn = this.getConnection();
		PreparedStatement statement = null;
		ResultSet resultSet = null;

		// Build SQL
		List<DatabaseField> fields = bean.getFields();
		String tableName = bean.getTableName();
		StringBuilder sb = new StringBuilder("INSERT INTO ")
			.append(tableName).append(" (");
		StringBuilder params = new StringBuilder(" (");
		for (DatabaseField field : fields) {
			
			if (!field.autoGenerate()) {
				// Only insert non generated fields
				sb.append(field.name()).append(",");
				params.append("?,");
			}
		}
		sb.deleteCharAt(sb.length() - 1).append(") VALUES ");
		params.deleteCharAt(params.length() - 1).append(") ");
		String sql = sb.append(params).toString();


		try {
			// Get the auto generated fields.
			List<DatabaseField> autoGeneratedFields = bean.getAutoGeneratedFields();
			String[] fieldsName = new String[autoGeneratedFields.size()];
			for (int i = 0; i < autoGeneratedFields.size(); i ++) {
				DatabaseField field = autoGeneratedFields.get(i);
				fieldsName[i] = field.name();
			}
			
			// Prepare parameters.
			statement = conn.prepareStatement(sql
					, Statement.RETURN_GENERATED_KEYS); //fieldsName); // 
			int index = 1;
			for (DatabaseField field : fields) {
				
				if (!field.autoGenerate()) {
					// Set parameter
					statement.setObject(index, bean.getFieldValue(field));
					index ++;
				}
			}
			
			// Execute SQL
			int rows = statement.executeUpdate();
			log("Insert executed: " + sql);
			
			// Note: this part is kind of tricky because this JDBC API is not well supported
			// By every database vender.
			
			// Populate the auto generated fields
			resultSet = statement.getGeneratedKeys();
	        if (null != resultSet && resultSet.next()) {
	        	
	        	index = 1;
				for (DatabaseField field : autoGeneratedFields) {
					
					// Get result value
					// The get by name method seems not working, tricky...
					//Object value = resultSet.getObject(field.name());
					Object value = resultSet.getObject(index);
					index ++;
					
					// Set bean fields
					bean.setFieldValue(field, value);
				}
	        }
	        
	        // Return the affected rows number.
			return rows;
			
		} catch (SQLException e) {
			String msg = "Insert failed: " + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} catch (InvalidFieldException e) {
			String msg = "Insert failed: " + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} finally {
			// Clean up
			if (resultSet != null) {
				try {
					resultSet.close();
					
				} catch (SQLException e) {
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			if (statement != null) {
				try {
					statement.close();
					
				} catch (SQLException e) {
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			this.closeConnection();
		}
	}

	
	/* (non-Javadoc)
	 * @see data.mysql.IDao#delete(data.bean.DataBean)
	 */
	@Override
	public int delete(Bean bean) throws DaoException {

		if (!this.isTableEntity()) {
			String msg = 
					"Please override this method to implement non table operation.";
			this.error(msg);
			throw new DaoException(msg);
		}
		
		// Get connection
		Connection conn = this.getConnection();
		PreparedStatement statement = null;

		// Build SQL
		String tableName = bean.getTableName();
		StringBuilder sb = new StringBuilder("DELETE FROM ")
			.append(tableName).append(" WHERE ");
		
		// Add where clause: by primary keys
		List<DatabaseField> fields = bean.getPrimaryKey();
		String logic = "";
		for (DatabaseField field : fields) {
			
			sb.append(logic).append(field.name()).append("=").append("?");
			logic = " AND ";
		}
		String sql = sb.toString();


		try {
			// Prepare parameters.
			statement = conn.prepareStatement(sql);
			int index = 1;
			for (DatabaseField field : fields) {
				
				// Set parameter
				statement.setObject(index, bean.getFieldValue(field));
				index ++;
			}
			
			// Execute SQL
			int rows = statement.executeUpdate();
			log("Delete executed: " + sql);
	        
	        // Return the affected rows number.
			return rows;
			
		} catch (SQLException e) {
			String msg = "Delete failed: " + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} catch (InvalidFieldException e) {
			String msg = "Delete failed: " + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} finally {
			// Clean up
			if (statement != null) {
				try {
					statement.close();
					
				} catch (SQLException e) {
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			this.closeConnection();
		}
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#deleteAll(data.bean.DataBean)
	 */
	@Override
	public int deleteAll() throws DaoException {

		if (!this.isTableEntity()) {
			String msg = 
					"Please override this method to implement non table operation.";
			this.error(msg);
			throw new DaoException(msg);
		}
		
		// Get connection
		Connection conn = this.getConnection();
		PreparedStatement statement = null;

		// Build SQL
		Bean bean = this.getBeanInstance();
		String tableName = bean.getTableName();
		StringBuilder sb = new StringBuilder(
				"DELETE FROM ").append(tableName);
		String sql = sb.toString();


		try {
			// Prepare parameters.
			statement = conn.prepareStatement(sql);
			
			// Execute SQL
			int rows = statement.executeUpdate();
			log("DeleteAll executed: " + sql);
	        
	        // Return the affected rows number.
			return rows;
			
		} catch (SQLException e) {
			String msg = "DeleteAll failed: " + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} finally {
			// Clean up
			if (statement != null) {
				try {
					statement.close();
					
				} catch (SQLException e) {
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			this.closeConnection();
		}
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#update(data.bean.DataBean)
	 */
	@Override
	public int update(Bean bean) throws DaoException {

		if (!this.isTableEntity()) {
			String msg = 
					"Please override this method to implement non table operation.";
			this.error(msg);
			throw new DaoException(msg);
		}
		
		// Get connection
		Connection conn = this.getConnection();
		PreparedStatement statement = null;

		// Build SQL
		String tableName = bean.getTableName();
		StringBuilder sb = new StringBuilder("UPDATE ")
			.append(tableName).append(" SET ");
		
		// SQL Update part.
		List<DatabaseField> fields = bean.getFields();
		String logic = "";
		for (DatabaseField field : fields) {
			
			if (!field.primaryKey()) {
				// Only update non primary key fields
				sb.append(logic).append(field.name()).append("=").append("?");
				logic = ", ";
			}
		}
		
		// Add where clause: by primary keys
		sb.append(" WHERE ");
		List<DatabaseField> pkFields = bean.getPrimaryKey();
		logic = "";
		for (DatabaseField field : pkFields) {
			
			sb.append(logic).append(field.name()).append("=").append("?");
			logic = " AND ";
		}
		String sql = sb.toString();


		try {
			// Prepare parameters.
			statement = conn.prepareStatement(sql);
			int index = 1;
			// Set parameters
			for (DatabaseField field : fields) {
				
				if (!field.primaryKey()) {
					statement.setObject(index, bean.getFieldValue(field));
					index ++;
				}
			}
			for (DatabaseField field : pkFields) {
				
				statement.setObject(index, bean.getFieldValue(field));
				index ++;
			}
			
			// Execute SQL
			int rows = statement.executeUpdate();
			log("Update executed: " + sql);
	        
	        // Return the affected rows number.
			return rows;
			
		} catch (SQLException e) {
			String msg = "Update failed: " + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} catch (InvalidFieldException e) {
			String msg = "Update failed: " + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} finally {
			// Clean up
			if (statement != null) {
				try {
					statement.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			this.closeConnection();
		}
	}

	/**
	 * Execute prepared update (insert, delete, update) SQL statement. 
	 * 
	 * @return 
	 * 			the number of rows affected.
	 * @throws DaoException 
	 * 			the DAO exception.
	 */
	protected int executeUpdate(
			String preparedSQL, Object... params) throws DaoException {

		// Get connection
		Connection conn = this.getConnection();
		PreparedStatement statement = null;

		try {
			// Prepare parameters.
			statement = conn.prepareStatement(preparedSQL);
			int index = 1;
			for (Object param : params) {

				// Set parameters
				statement.setObject(index, param);
				index ++;
			}
			
			// Execute SQL
			int rows = statement.executeUpdate();
			log("executeUpdate executed: " + preparedSQL);
	        
	        // Return the affected rows number.
			return rows;
			
		} catch (SQLException e) {
			String msg = "executeUpdate failed: " + preparedSQL;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} finally {
			// Clean up
			if (statement != null) {
				try {
					statement.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			this.closeConnection();
		}
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#selectByField(java.lang.String, java.lang.Object, data.Pagination)
	 */
	@Override
	public List<Bean> selectByField(
			String fieldName, Object value, Pagination pagination) throws DaoException {

		if (!this.isTableEntity()) {
			String msg = 
					"Please override this method to implement non table operation.";
			this.error(msg);
			throw new DaoException(msg);
		}
		
		// Get connection
		Connection conn = this.getConnection();
		PreparedStatement statement = null;
		ResultSet resultSet = null;

		// Build select clause
		Bean bean = this.getBeanInstance();
		String tableName = bean.getTableName();
		List<DatabaseField> fields = bean.getFields();
		String sql = buildSelectClause(fields, tableName);
		
		// Build where clause
		StringBuilder sb = new StringBuilder(sql);
		sql = sb.append(" WHERE ").append(fieldName).append("=?").toString();
		
		// Add pagination support
		if (pagination != null) {
			sql = this.addPagination(sql, pagination);
		}
		
		try {

			// Prepare statement
			statement = conn.prepareStatement(sql);
			statement.setObject(1, value);
			
			// Execute query
			resultSet = statement.executeQuery();
			log("SelectByField executed: " + sql);
			
			// Populate and return results.
			List<Bean> results = populateResults(resultSet, fields);

			// Return results
			return results;
			
		} catch (SQLException e) {
			String msg = "SelectByField failed:" + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} catch (DaoException e) {
			String msg = "SelectByField failed:" + sql;
			error(msg, e);
			throw e;
		
		} finally {
			// Clean up
			if (resultSet != null) {
				try {
					resultSet.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			if (statement != null) {
				try {
					statement.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			this.closeConnection();
		}
	}
	
	/* (non-Javadoc)
	 * @see data.mysql.IDao#selectByField(java.lang.String, java.lang.Object)
	 */
	@Override
	public List<Bean> selectByField(
			String fieldName, Object value) throws DaoException {

		// Select without pagination
		return this.selectByField(fieldName, value, null);
	}
	

	/* (non-Javadoc)
	 * @see data.mysql.IDao#selectByPrimaryKey(java.util.Map<String, Object>)
	 */
	@Override
	public Bean selectByPrimaryKey(Map<String, Object> pk) throws DaoException {

		if (!this.isTableEntity()) {
			String msg = 
					"Please override this method to implement non table operation.";
			this.error(msg);
			throw new DaoException(msg);
		}
		
		// Validate input
		Bean bean = this.getBeanInstance();
		List<DatabaseField> pkfields = bean.getPrimaryKey();
		if (pk.size() != pkfields.size()) {

			String msg = "Invalid primary key input for selectByPrimaryKey method";
			error(msg);
			throw new DaoException(msg);
		}
		Map<String, Object> pkValues = pk;
		
		// Get connection
		Connection conn = this.getConnection();
		PreparedStatement statement = null;
		ResultSet resultSet = null;

		// Build select clause
		String tableName = bean.getTableName();
		List<DatabaseField> fields = bean.getFields();
		String sql = buildSelectClause(fields, tableName);
		
		// Build where clause
		StringBuilder sb = new StringBuilder(sql);
		sb.append(" WHERE ");
		String logic = "";
		for (DatabaseField field : pkfields) {
			
			sb.append(logic).append(field.name()).append("=").append("?");
			logic = " AND ";
		}
		sql = sb.toString();
		
		try {
			// Prepare statement
			statement = conn.prepareStatement(sql);
			int index = 1;
			for (DatabaseField field : pkfields) {

				// Set parameters
				String fieldName = field.name();
				if (!pkValues.containsKey(fieldName)) {
					String msg = 
							"Invalid primary key input for selectByPrimaryKey method: no field " 
							+ fieldName;
					error(msg);
					throw new DaoException(msg);
				}
				Object value = pkValues.get(fieldName);
				statement.setObject(index, value);
				index ++;
			}

			// Execute query
			resultSet = statement.executeQuery();
			log("selectByPrimaryKey executed: " + sql);
			
			// Populate and return results.
			List<Bean> results = populateResults(resultSet, fields);
			
			// The result should be null or unique since it is primary key
			if (results == null || results.size() != 1) {
				
				// Not found.
				return null;
			}

			// Return results
			return results.get(0);
			
		} catch (SQLException e) {
			String msg = "selectByPrimaryKey failed:" + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} catch (DaoException e) {
			String msg = "selectByPrimaryKey failed:" + sql;
			error(msg, e);
			throw e;
		
		} finally {
			// Clean up
			if (resultSet != null) {
				try {
					resultSet.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			if (statement != null) {
				try {
					statement.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			this.closeConnection();
		}
	}

	/* (non-Javadoc)
	 * @see data.mysql.IDao#selectAll(dataPagination)
	 */
	@Override
	public List<Bean> selectAll(Pagination pagination) throws DaoException {

		if (!this.isTableEntity()) {
			String msg = 
					"Please override this method to implement non table operation.";
			this.error(msg);
			throw new DaoException(msg);
		}
		
		// Get connection
		Connection conn = this.getConnection();
		Statement statement = null;
		ResultSet resultSet = null;
	
		// Build query
		Bean bean = this.getBeanInstance();
		List<DatabaseField> fields = bean.getFields();
		String tableName = bean.getTableName();
		String sql = buildSelectClause(fields, tableName);
		
		// Add pagination support
		if (pagination != null) {
			
			sql = this.addPagination(sql, pagination);
		}
		
		try {
			// Execute query
			statement = conn.createStatement();
			resultSet = statement.executeQuery(sql);
			log("SelectAll executed: " + sql);
			
			// Populate and return results.
			List<Bean> results = populateResults(resultSet, bean.getFields());
			
			// Return results.
			return results;
			
		} catch (SQLException e) {
			String msg = "SelectAll failed:" + sql;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} catch (DaoException e) {
			String msg = "SelectAll failed:" + sql;
			error(msg, e);
			throw e;
		
		} finally {
			// Clean up
			if (resultSet != null) {
				try {
					resultSet.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			if (statement != null) {
				try {
					statement.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			this.closeConnection();
		}
	}
	
	/* (non-Javadoc)
	 * @see data.mysql.IDao#selectAll()
	 */
	@Override
 	public List<Bean> selectAll() throws DaoException {
		
		// Select without pagination
		return this.selectAll(null);
	}

	/**
	 * Build the SELECT FROM clause in query SQL.
	 * @param fields the fields to select.
	 * @param tableName the table name.
	 * @return the SQL clause.
	 */
	protected String buildSelectClause(
			List<DatabaseField> fields, String tableName) {
		
		// SELECT clause
		StringBuilder sb = new StringBuilder("SELECT ");
		int size = fields.size();
		for (int i = 0; i < size; i++) {
			
			DatabaseField field = fields.get(i);
			sb.append(field.name());
			if (i < size - 1) {
				sb.append(", ");
			}
		}
		
		// FROM clause
		sb.append(" FROM ").append(tableName).append(" ");
		return sb.toString();
	}

	/**
	 * Build the SELECT FROM clause in query SQL.
	 * @param fields the fields to select.
	 * @param tableName the table name.
	 * @return the SQL clause.
	 */
	protected String buildSelectClause(
			String[] fields, String tableName) {
		
		// SELECT clause
		StringBuilder sb = new StringBuilder("SELECT ");
		int size = fields.length;
		for (int i = 0; i < size; i++) {
			
			sb.append(fields[i]);
			if (i < size - 1) {
				sb.append(", ");
			}
		}
		
		// FROM clause
		sb.append(" FROM ").append(tableName).append(" ");
		return sb.toString();
	}

	/**
	 * Populate result bean from result set.
	 */
	public List<Bean> populateResults(
			ResultSet resultSet, List<DatabaseField> fields) throws DaoException {
		
		List<Bean> results = new ArrayList<Bean>();
		try {
			
			while (resultSet.next()) {
				
				// Add to results
				Bean result = this.createBean();
				results.add(result);
				for (DatabaseField field : result.getFields()) {
					
					// Get result value
					Object value = resultSet.getObject(field.name());
					//Object value = resultSet.getObject(field.name(), field.type());
	
					// Set bean fields	
					result.setFieldValue(field, value);
				}
	        }
			
		} catch (InvalidFieldException e) {
			String msg = "Failed to populate results: invalid field!";
			logger.error(msg, e);
			
		} catch (SQLException e) {
			String msg = "Failed to populate results.";
			logger.error(msg, e);
		}
		return results;
	}

	/* (non-Javadoc)
	 * @see data.mysql.IResultProcessor#populateResults(
	 * 		java.sql.ResultSet, java.lang.String[])
	 */
	@Override
	public List<Bean> populateResults(
			ResultSet resultSet, String[] fieldsName) throws DaoException {
		
		List<DatabaseField> fields = this.getBeanInstance().getFields(
				new ArrayList<String>(Arrays.asList(fieldsName)));
		return this.populateResults(resultSet, fields);
	}
	
	/**
	 * Execute prepared query SQL.
	 */
	protected <ResultType> ResultType executeQuery(
			IResultsProcessor<ResultType> processor, List<DatabaseField> fields,
			String preparedSQL, Object... params) throws DaoException {
		
		// Execute query without pagination
		return this.executeQuery(processor, fields, null, preparedSQL, params);
	}

	/**
	 * Execute prepared query SQL.
	 */
	protected <ResultType> ResultType executeQuery(
			IResultsProcessor<ResultType> processor, List<DatabaseField> fields,
			Pagination pagination,
			String preparedSQL, Object... params) throws DaoException {
		
		// Builde fields name array
		String[] fieldsName = new String[fields.size()];
		for (int index = 0; index < fields.size(); index++) {
			DatabaseField field = fields.get(index);
			fieldsName[index] = field.name();
		}
		
		return this.executeQuery(processor, fieldsName, pagination, preparedSQL, params);
	}
	
	/**
	 * Execute prepared query SQL.
	 */
	protected <ResultType> ResultType executeQuery(
			IResultsProcessor<ResultType> processor, String[] fieldsName,
			String preparedSQL, Object... params) throws DaoException {
		
		// Execute query without pagination
		return this.executeQuery(processor, fieldsName, null, preparedSQL, params);
	}

	/**
	 * Execute prepared query SQL with pagination.
	 */
	protected <ResultType> ResultType executeQuery(
			IResultsProcessor<ResultType> processor, String[] fieldsName, 
			Pagination pagination, 
			String preparedSQL, Object... params) throws DaoException {

		// Get connection
		Connection conn = this.getConnection();
		PreparedStatement statement = null;
		ResultSet resultSet = null;
		
		// Add pagination support
		if (pagination != null) {
			
			preparedSQL = this.addPagination(preparedSQL, pagination);
		}

		try {
			// Prepare parameters.
			statement = conn.prepareStatement(preparedSQL);
			int index = 1;
			for (Object param : params) {

				// Set parameters
				statement.setObject(index, param);
				index ++;
			}
			
			// Execute SQL
			resultSet = statement.executeQuery();
			log("executeQuery executed: " + preparedSQL);
	        
			// Process result
			return processor.populateResults(resultSet, fieldsName);
			
		} catch (SQLException e) {
			
			String msg = "executeQuery failed: " + preparedSQL;
			error(msg, e);
			throw new DaoException(msg, e);
			
		} catch (DaoException e) {
			
			String msg = 
					"executeQuery failed, failed to populate results: "+ preparedSQL;
			error(msg, e);
			throw e;
			
		} finally {
			// Clean up
			if (resultSet != null) {
				try {
					resultSet.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			if (statement != null) {
				try {
					statement.close();
					
				} catch (SQLException e) {
					
					e.printStackTrace();
					throw new DaoException(e);
				}
			}
			this.closeConnection();
		}
	}
	
	/**
	 * Add pagination to the original query statement.
	 * 
	 * @param querySQL the query SQL without pagination.
	 * @param pagination the pagination information.
	 * 
	 * @return the query SQL with pagination support.
	 */
	protected String addPagination(String querySQL, Pagination pagination) {
		
		// Offset and row count
		int rowCount = pagination.getPageSize();
		int pageNo = pagination.getPageNo();	// Start from 1
		int offset = (pageNo - 1) * rowCount;
		
		// MySQL supports optimized pagination
		// The offset of the initial row is 0 (not 1)
		StringBuilder sql = new StringBuilder(querySQL);
		sql.append(" LIMIT ").append(
				offset).append(",").append(rowCount).toString();
		return sql.toString();
	}
}

